# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time
from jms.dm.dm_customer_loss_warning_month_m import jms_dm__dm_customer_loss_warning_month_m
from utils.operators.rich_sql_sensor import RichSqlSensor

label = f"jms_dwd_customer_loss_warning_month_{int(time())}"
timeout = timedelta(minutes=30).seconds
doris_jms_dwd__dwd_customer_loss_warning_month = RichSqlSensor(
    task_id='doris_jms_dwd__dwd_customer_loss_warning_month',
    pool='broker_load_pool',
    email=['guoruiling@jtexpress.com','yl_bigdata@yl-scm.com'],
    conn_id='doris',
    pre_sql=f"""
    TRUNCATE TABLE jms_dwd.dwd_customer_loss_warning_month PARTITION (p{{{{ execution_date | cst_month_nodash }}}});
    LOAD LABEL jms_dwd.{label} (
        DATA INFILE(
                    "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_customer_loss_warning_month_m/mt={{{{ execution_date | cst_month_dash }}}}/*/*"
                    )
        INTO TABLE dwd_customer_loss_warning_month  
    FORMAT AS 'PARQUET'
    (network_level, date_time, beginning_month_date, agent_name, agent_code, fran_name, fran_code, network_name, network_code, customer_level, customer_cnt, loss_customer_cnt, loss_customer_ticket_qty, loss_customer_rt, save_customer_cnt, save_customer_ticket_qty, save_customer_rt, new_customer_cnt, new_customer_ticket_qty, new_customer_rt, warning_customer_cnt, warning_customer_ticket_qty, warning_customer_cnt_up, warning_customer_cnt_ring_rt)
     )
    WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
    PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_dwd WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dwd WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120), )

doris_jms_dwd__dwd_customer_loss_warning_month << jms_dm__dm_customer_loss_warning_month_m
